#pragma once
#ifndef __linux__
#error unix local stream only support linux system
#endif

#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <fcntl.h>
#include <unistd.h>
#include <string.h>
#include <stddef.h>
#ifndef INVALID_SOCKET
#define INVALID_SOCKET -1
#endif
#include "com.h"

namespace KRPC {

class LOCALSTREAM : public COM
{
private:
	static socklen_t size(const struct sockaddr_un &addr)
	{
		return offsetof(struct sockaddr_un, sun_path) + strlen(addr.sun_path);
	}
	void setRandomAddr(struct sockaddr_un &addr)
	{
		char softlink[40] = {0}, exePath[128] = {0};
		sprintf(softlink, "/proc/%d/exe", getpid());
		int count = readlink(softlink, exePath, sizeof(exePath));
		assert(count > 0 && count <= (int)(sizeof(exePath)));
		const char *name = strrchr(exePath, '/');
		name = (name == nullptr) ? exePath : (name + 1);	
		assert(strlen(name) + 30 <= sizeof(addr.sun_path));
		//前11个字节填'/tmp/local/',中间填name,后面17个字节填'_6位整数pid_stream4位整数sockfd',最后1字节填'\0'
		struct stat st;
		if (stat("/tmp/local", &st) == -1 || S_ISDIR(st.st_mode) == 0) {
			(void) mkdir("/tmp/local", 0777);	//析构函数中增加了删除逻辑，不需要依赖其他进程去清除
		}
		// tmp 目录重启后会清空 进程名+pid+协议类型+socketfd , 在同一系统同一时刻具有唯一性
		int n = sprintf(addr.sun_path, "/tmp/local/%s_%d_stream%d", name, getpid(), sockfd);
		addr.sun_path[n] = '\0';
		if (access(addr.sun_path, F_OK) == 0)
			unlink(addr.sun_path);
		addr.sun_family = AF_LOCAL;
	}
public:
	LOCALSTREAM(const char *filename)
	{
		sockfd = INVALID_SOCKET;
		dstaddr.sun_family = AF_LOCAL;
		memset(dstaddr.sun_path, 0, sizeof(dstaddr.sun_path));
		std::string path = "/tmp/local/";
		path += filename;
		strcpy(dstaddr.sun_path, path.c_str());
		open();
	}

	~LOCALSTREAM()
	{
		close();
	}

	void open()
	{
		if (sockfd == INVALID_SOCKET) {
			sockfd = socket(AF_LOCAL, SOCK_STREAM, 0);
			setRandomAddr(srcaddr);
			connected = false;
			(void) bind(sockfd, (struct sockaddr *)&srcaddr, size(srcaddr));
		}
	}

	void close()
	{
		if (sockfd == INVALID_SOCKET)
			return;
		::close(sockfd);
		sockfd = INVALID_SOCKET;
		connected = false;
		unlink(srcaddr.sun_path); 
	}

	bool connect(const struct sockaddr_un &addr)
	{
		if (!connected) {
			open();
#if 0 //阻塞连接
			connected = (::connect(sockfd, (const struct sockaddr *)&addr, size(addr)) == 0);
#else // 超时连接
			int flags = fcntl(sockfd, F_GETFL, 0);
			if(-1 == fcntl(sockfd, F_SETFL, flags | O_NONBLOCK))
			{
				close();
				return false;
			}
    		if (::connect(sockfd, (const struct sockaddr *)&addr, size(addr)) == -1) {
    			if (errno != EINPROGRESS) {
    				close();
    				return false;
    			} else {
    				fd_set fdw;
    				timeval timeout = {2, 0};	//默认超时2s
    				FD_ZERO(&fdw);
					FD_SET(sockfd, &fdw);
					if (select(sockfd + 1, NULL, &fdw, NULL, &timeout) <= 0) {
	    				close();
	    				return false;
	    			}
	    			if (::connect(sockfd, (const struct sockaddr *)&addr, size(addr)) == 0 || errno == EISCONN) {
	    				connected = true;
	    			} else {
	    				close();
    					return false;
	    			}
    			}
    		} else {
    			connected = true;
    		}
			if (-1 == fcntl(sockfd, F_SETFL, flags)){/*fix coverity*/}
#endif
		}
		return connected;
	}

	virtual bool recv(Packet &packet, int msec) override
	{
		if (!connect(dstaddr))
			return false;
		setRcvTimeout(msec);
		MsgHead head;
		auto rcvsz = ::recv(sockfd, (char *)&head, sizeof(MsgHead), 0);
		auto RecvDeal = [&rcvsz, this]()->bool{
			if (rcvsz < 0) {
				if (errno != EWOULDBLOCK) {
					close();	//如果不是由于超时引起的读取失败则关闭连接
				} 
				return false;
			} else if (rcvsz == 0) {
				close();
				return false;
			} else {
				return true;
			}
		};

		if (!RecvDeal() || rcvsz != sizeof(MsgHead) || head.len < sizeof(MsgHead)) 
			return false;
		packet.resize(head.len);
		packet.write((const char *)&head, sizeof(MsgHead));

		auto leftsz = head.len - sizeof(MsgHead);
		std::size_t offset = sizeof(MsgHead);
		while (leftsz > 0) {
			rcvsz = ::recv(sockfd, (char *)packet.data() + offset, leftsz, 0);
			if (!RecvDeal()) 
				return false;
			offset += rcvsz;
			leftsz -= rcvsz;
			packet.reset(offset);
		}
		return true;
	}

	void setRcvTimeout(int msec)
	{
		struct timeval tv = {msec / 1000, 1000 * (msec % 1000)};
		if (-1 == setsockopt(sockfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv))){/*fix coverity*/}
	}

	virtual bool send(const Packet &packet) override
	{
		auto sndsz = (ssize_t)packet.size();
		bool ret = connect(dstaddr) ? (::send(sockfd, packet.data(), sndsz, 0) == sndsz) : false;
		if (!ret) 
			close();
		return ret;
	}

private:
	int sockfd;
	bool connected;
	struct sockaddr_un srcaddr;
	struct sockaddr_un dstaddr;
};

}
